/**
 * @author 
 * @desc 读取指定开始与结束字符串中间的内容，当然，该类只适用于西维尔日志，因为是其中所用到
 * 		的byteQueue是根据日志格式读取其请求的时间，你也可以把byteQueue相关去掉来把该类作
 * 		为通类来适用
 * 
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.ccb.recordreader;

import java.io.Closeable;
import java.io.IOException;
import java.io.InputStream;

import org.apache.hadoop.classification.InterfaceAudience;
import org.apache.hadoop.classification.InterfaceStability;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.Text;

//import com.maywide.util.ByteQueue;
/**
 * A class that provides a line reader from an input stream. Depending on the constructor used, lines will either be terminated by:
 * <ul>
 * <li>one of the following: '\n' (LF) , '\r' (CR), or '\r\n' (CR+LF).</li>
 * <li><em>or</em>, a custom byte sequence delimiter</li>
 * </ul>
 * In both cases, EOF also terminates an otherwise unterminated line.
 */
@InterfaceAudience.LimitedPrivate({ "MapReduce" })
@InterfaceStability.Unstable
public class CustomLineReader implements Closeable {

	private static final int DEFAULT_BUFFER_SIZE = 64 * 1024;
	private int bufferSize = DEFAULT_BUFFER_SIZE;
	private InputStream in;
	private byte[] buffer;
	// the number of bytes of real data in the buffer
	private int bufferLength = 0;
	// the current position in the buffer
	private int bufferPosn = 0;

	private static final byte CR = '\r';
	private static final byte LF = '\n';

	// 得到开始截断的位置前读过多少bytes
	private long startMarkbytesConsumed = 0;
	// The line delimiter
	private final byte[] startRecordDelimiterBytes;
	private final byte[] endRecordDelimiterBytes;

	// 存储startRecordDelimiterBytes所读过的字符流
	// private ByteQueue byteQueue = new ByteQueue(800);

	/**
	 * Create a line reader that reads from the given stream using the default buffer-size (64k).
	 * 
	 * @param in
	 *            The input stream
	 * @throws IOException
	 */
	public CustomLineReader(InputStream in) {
		this(in, DEFAULT_BUFFER_SIZE);
	}

	/**
	 * Create a line reader that reads from the given stream using the given buffer-size.
	 * 
	 * @param in
	 *            The input stream
	 * @param bufferSize
	 *            Size of the read buffer
	 * @throws IOException
	 */
	public CustomLineReader(InputStream in, int bufferSize) {
		this.in = in;
		this.bufferSize = bufferSize;
		this.buffer = new byte[this.bufferSize];
		this.endRecordDelimiterBytes = null;
		this.startRecordDelimiterBytes = null;
	}

	/**
	 * Create a line reader that reads from the given stream using the <code>io.file.buffer.size</code> specified in the given <code>Configuration</code>.
	 * 
	 * @param in
	 *            input stream
	 * @param conf
	 *            configuration
	 * @throws IOException
	 */
	public CustomLineReader(InputStream in, Configuration conf) throws IOException {
		this(in, conf.getInt("io.file.buffer.size", DEFAULT_BUFFER_SIZE));
	}

	/**
	 * Create a line reader that reads from the given stream using the default buffer-size, and using a custom delimiter of array of bytes.
	 * 
	 * @param in
	 *            The input stream
	 * @param recordDelimiterBytes
	 *            The delimiter
	 */
	public CustomLineReader(InputStream in, byte[] startRecordDelimiterBytes, byte[] recordDelimiterBytes) {
		this.in = in;
		this.bufferSize = DEFAULT_BUFFER_SIZE;
		this.buffer = new byte[this.bufferSize];
		this.endRecordDelimiterBytes = recordDelimiterBytes;
		this.startRecordDelimiterBytes = startRecordDelimiterBytes;
	}

	/**
	 * Create a line reader that reads from the given stream using the given buffer-size, and using a custom delimiter of array of bytes.
	 * 
	 * @param in
	 *            The input stream
	 * @param bufferSize
	 *            Size of the read buffer
	 * @param recordDelimiterBytes
	 *            The delimiter
	 * @throws IOException
	 */
	public CustomLineReader(InputStream in, int bufferSize, byte[] startRecordDelimiterBytes, byte[] recordDelimiterBytes) {
		this.in = in;
		this.bufferSize = bufferSize;
		this.buffer = new byte[this.bufferSize];
		this.endRecordDelimiterBytes = recordDelimiterBytes;
		this.startRecordDelimiterBytes = startRecordDelimiterBytes;
	}

	/**
	 * Create a line reader that reads from the given stream using the <code>io.file.buffer.size</code> specified in the given <code>Configuration</code>, and using a custom delimiter of array of bytes.
	 * 
	 * @param in
	 *            input stream
	 * @param conf
	 *            configuration
	 * @param recordDelimiterBytes
	 *            The delimiter
	 * @throws IOException
	 */
	public CustomLineReader(InputStream in, Configuration conf, byte[] startRecordDelimiterBytes, byte[] recordDelimiterBytes) throws IOException {
		this.in = in;
		this.bufferSize = conf.getInt("io.file.buffer.size", DEFAULT_BUFFER_SIZE);
		this.buffer = new byte[this.bufferSize];
		this.endRecordDelimiterBytes = recordDelimiterBytes;
		this.startRecordDelimiterBytes = startRecordDelimiterBytes;
	}

	/**
	 * Close the underlying stream.
	 * 
	 * @throws IOException
	 */
	public void close() throws IOException {
		in.close();
	}

	/**
	 * Read one line from the InputStream into the given Text.
	 *
	 * @param str
	 *            the object to store the given line (without newline)
	 * @param maxLineLength
	 *            the maximum number of bytes to store into str; the rest of the line is silently discarded.
	 * @param maxBytesToConsume
	 *            the maximum number of bytes to consume in this call. This is only a hint, because if the line cross this threshold, we allow it to happen. It can overshoot potentially by as much as one buffer length.
	 *
	 * @return the number of bytes read including the (longest) newline found.
	 *
	 * @throws IOException
	 *             if the underlying stream throws
	 */
	public int readLine(Text str, int maxLineLength, int maxBytesToConsume) throws IOException {
		if (this.endRecordDelimiterBytes != null) {
			return readCustomLine(str, maxLineLength, maxBytesToConsume);
		} else {
			return readDefaultLine(str, maxLineLength, maxBytesToConsume);
		}
	}

	protected int fillBuffer(InputStream in, byte[] buffer, boolean inDelimiter) throws IOException {
		return in.read(buffer);
	}

	/**
	 * Read a line terminated by one of CR, LF, or CRLF.
	 */
	private int readDefaultLine(Text str, int maxLineLength, int maxBytesToConsume) throws IOException {
		/*
		 * We're reading data from in, but the head of the stream may be already buffered in buffer, so we have several cases: 1. No newline characters are in the buffer, so we need to copy everything and read another buffer from the stream. 2. An unambiguously terminated line is in buffer, so we just copy to str. 3. Ambiguously terminated line is in buffer, i.e. buffer ends in CR. In this case we copy everything up to CR to str, but we also need to see what follows CR: if it's LF, then we need consume LF as well, so next call to readLine will read from after that. We use a flag prevCharCR to signal if previous character was CR and, if it happens to be at the end of the buffer, delay consuming it until we have a chance to look at the char that follows.
		 */
		str.clear();
		int txtLength = 0; // tracks str.getLength(), as an optimization
		int newlineLength = 0; // length of terminating newline
		boolean prevCharCR = false; // true of prev char was CR
		long bytesConsumed = 0;
		do {
			int startPosn = bufferPosn; // starting from where we left off the last time
			if (bufferPosn >= bufferLength) {
				startPosn = bufferPosn = 0;
				if (prevCharCR) {
					++bytesConsumed; // account for CR from previous read
				}
				bufferLength = fillBuffer(in, buffer, prevCharCR);
				if (bufferLength <= 0) {
					break; // EOF
				}
			}
			for (; bufferPosn < bufferLength; ++bufferPosn) { // search for newline
				if (buffer[bufferPosn] == LF) {
					newlineLength = (prevCharCR) ? 2 : 1;
					++bufferPosn; // at next invocation proceed from following byte
					break;
				}
				if (prevCharCR) { // CR + notLF, we are at notLF
					newlineLength = 1;
					break;
				}
				prevCharCR = (buffer[bufferPosn] == CR);
			}
			int readLength = bufferPosn - startPosn;
			if (prevCharCR && newlineLength == 0) {
				--readLength; // CR at the end of the buffer
			}
			bytesConsumed += readLength;
			int appendLength = readLength - newlineLength;
			if (appendLength > maxLineLength - txtLength) {
				appendLength = maxLineLength - txtLength;
			}
			if (appendLength > 0) {
				str.append(buffer, startPosn, appendLength);
				txtLength += appendLength;
			}
		} while (newlineLength == 0 && bytesConsumed < maxBytesToConsume);

		if (bytesConsumed > Integer.MAX_VALUE) {
			throw new IOException("Too many bytes before newline: " + bytesConsumed);
		}
		return (int) bytesConsumed;
	}

	/**
	 * Read a line terminated by a custom delimiter.
	 */
	private int readCustomLine(Text str, int maxLineLength, int maxBytesToConsume) throws IOException {
		/*
		 * We're reading data from inputStream, but the head of the stream may be already captured in the previous buffer, so we have several cases:
		 * 
		 * 1. The buffer tail does not contain any character sequence which matches with the head of delimiter. We count it as a ambiguous byte count = 0
		 * 
		 * 2. The buffer tail contains a X number of characters, that forms a sequence, which matches with the head of delimiter. We count ambiguous byte count = X
		 * 
		 * // *** eg: A segment of input file is as follows
		 * 
		 * " record 1792: I found this bug very interesting and I have completely read about it. record 1793: This bug can be solved easily record 1794: This ."
		 * 
		 * delimiter = "record";
		 * 
		 * supposing:- String at the end of buffer = "I found this bug very interesting and I have completely re" There for next buffer = "ad about it. record 179       ...."
		 * 
		 * The matching characters in the input buffer tail and delimiter head = "re" Therefore, ambiguous byte count = 2 **** //
		 * 
		 * 2.1 If the following bytes are the remaining characters of the delimiter, then we have to capture only up to the starting position of delimiter. That means, we need not include the ambiguous characters in str.
		 * 
		 * 2.2 If the following bytes are not the remaining characters of the delimiter ( as mentioned in the example ), then we have to include the ambiguous characters in str.
		 */
		int txtLength = 0; // tracks str.getLength(), as an optimization
		long bytesConsumed = 0;// record的长度
		int delPosn = 0;// //记录当前已匹配到delimiter的第几个字符
		int ambiguousByteCount = 0; // To capture the ambiguous characters count //保存上一次buffer结束时候匹配到结束字符串的第几个字符
		str.clear();
		// 这个是我加上去的，可以限定record从哪里开始哦
		this.startMarkbytesConsumed = 0;
		bufferPosn = readStartMark(this.startRecordDelimiterBytes);
		bytesConsumed += this.startMarkbytesConsumed;
		// 添加匹配头
		str.append(this.startRecordDelimiterBytes, 0, this.startRecordDelimiterBytes.length);

		do {
			int startPosn = bufferPosn; // Start from previous end position
			if (bufferPosn >= bufferLength) {// 貌似读到了当前buffer尽头了
				startPosn = bufferPosn = 0;
				// 又要重新去load缓存的数据
				bufferLength = fillBuffer(in, buffer, ambiguousByteCount > 0);
				if (bufferLength <= 0) {// 如果未能load进数据，要把那匹配的几个delimiter中的字符加上去
					str.append(endRecordDelimiterBytes, 0, ambiguousByteCount);

					break; // EOF
				}
			}
			for (; bufferPosn < bufferLength; ++bufferPosn) {
				// System.out.println((char)buffer[bufferPosn]);
				if (buffer[bufferPosn] == endRecordDelimiterBytes[delPosn]) {// 又找到一个匹配的字符了
					// 记录当前已匹配到delimiter的第几个字符
					delPosn++;
					if (delPosn >= endRecordDelimiterBytes.length) {// 恭喜你
						// 完全可以匹配到delimiter，跳出循环，就要返回了。
						bufferPosn++;
						break;
					}
				}
				// 很不幸，未能完全匹配到delimiter
				else if (delPosn != 0) {
					bufferPosn--;// 稍微前移一下
					delPosn = 0;
				}
			}
			int readLength = bufferPosn - startPosn;
			bytesConsumed += readLength;
			// 当重新读取下一个buffer时，先让str不添加delimiter的前部分
			int appendLength = readLength - delPosn;
			// 超过最大长度了，要截断
			if (appendLength > maxLineLength - txtLength) {
				appendLength = maxLineLength - txtLength;
			}
			if (appendLength > 0) {
				if (ambiguousByteCount > 0) {// 若新的buffer不能构成完整的delimiter，就应该增加上次buffer的尾巴
					str.append(endRecordDelimiterBytes, 0, ambiguousByteCount);
					// appending the ambiguous characters (refer case 2.2)
					// bytesConsumed增加上次buffer的尾巴的计数
					bytesConsumed += ambiguousByteCount;
					ambiguousByteCount = 0;
				}
				str.append(buffer, startPosn, appendLength);
				txtLength += appendLength;
			}
			if (bufferPosn >= bufferLength) {
				// 终于到一个buffer的终点了
				// 居然buffer的结尾匹配到了几个delimiter中的字符
				if (delPosn > 0 && delPosn < endRecordDelimiterBytes.length) {
					// bytesConsumed要先减去那几个字符
					ambiguousByteCount = delPosn;
					bytesConsumed -= ambiguousByteCount; // to be consumed in next
				}
			}
		} while (delPosn < endRecordDelimiterBytes.length && bytesConsumed < maxBytesToConsume);// 一次循环下来未能完全匹配一个结束字符串
		if (bytesConsumed > Integer.MAX_VALUE) {
			throw new IOException("Too many bytes before delimiter: " + bytesConsumed);
		}

		// 只有构成一个完整的开始与结束
		if (delPosn == endRecordDelimiterBytes.length) {
			// 作为结果返回
			str.append(this.endRecordDelimiterBytes, 0, this.endRecordDelimiterBytes.length);
			str.set(str.toString());
		}

		return (int) bytesConsumed;

	}

	private int readStartMark(byte[] startMarkBytes) throws IOException {
		// 记录当前bufferPosn下已经匹配了结束字符串的多少个字符
		int delPosn = 0;
		int ambiguousByteCount = 0; // To capture the ambiguous characters count 保存上一次buffer结束时候匹配到结束字符串的第几个字符
		do {
			int startPosn = bufferPosn; // Start from previous end position
			if (bufferPosn >= bufferLength) {// 哎呦，又要重新去load缓存的数据了
				startPosn = bufferPosn = 0;
				bufferLength = fillBuffer(in, buffer, ambiguousByteCount > 0);// /ambiguousByteCount > 0条件貌似，没用啊
				if (bufferLength <= 0) {
					break; // EOF
				}
			}
			for (; bufferPosn < bufferLength; ++bufferPosn) {
				// byteQueue.add(buffer[bufferPosn]);
				if (buffer[bufferPosn] == startMarkBytes[delPosn]) {
					// 记录当前已读到结束字符串的第几个字符
					delPosn++;
					// 完全可以匹配到结束字符串，跳出循环，就要返回了。
					if (delPosn >= startMarkBytes.length) {
						// byteQueue.revoke(1);
						bufferPosn++;
						break;
					}
				}
				// 很不幸，未能完全匹配到结束字符串，继续加油
				else if (delPosn != 0) {
					bufferPosn--;
					delPosn = 0;
				}
			}
			int readLength = bufferPosn - startPosn;
			this.startMarkbytesConsumed += readLength;
			// 当buffer重新读取流，让str不添加结束字符串的一部分
			int appendLength = readLength - delPosn;
			if (appendLength > 0) {
				if (ambiguousByteCount > 0) {
					// appending the ambiguous characters (refer case 2.2)
					ambiguousByteCount = 0;
				}
			}

			if (bufferPosn >= bufferLength) {
				// 终于到一个buffer的终点了
				// 那么，问题就来了，该死的，居然buffer的结尾匹配到了几个结束字符串中的字符
				if (delPosn > 0 && delPosn < startMarkBytes.length) {
					ambiguousByteCount = delPosn;
					// 所以，当前buffer只消耗了bytesConsumed
					this.startMarkbytesConsumed -= ambiguousByteCount; // to be consumed in next
				}
			}
		} while (delPosn < startMarkBytes.length);// 一次循环下来居然未能完全匹配一个结束字符串，再来一次

		return bufferPosn;
	}

	/**
	 * Read from the InputStream into the given Text.
	 * 
	 * @param str
	 *            the object to store the given line
	 * @param maxLineLength
	 *            the maximum number of bytes to store into str.
	 * @return the number of bytes read including the newline
	 * @throws IOException
	 *             if the underlying stream throws
	 */
	public int readLine(Text str, int maxLineLength) throws IOException {
		return readLine(str, maxLineLength, Integer.MAX_VALUE);
	}

	/**
	 * Read from the InputStream into the given Text.
	 * 
	 * @param str
	 *            the object to store the given line
	 * @return the number of bytes read including the newline
	 * @throws IOException
	 *             if the underlying stream throws
	 */
	public int readLine(Text str) throws IOException {
		return readLine(str, Integer.MAX_VALUE, Integer.MAX_VALUE);
	}
}
